package databus.client.consumer;

import com.google.protobuf.InvalidProtocolBufferException;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.databus.client.*;
import studio.raptor.databus.command.Event;

import javax.jms.JMSException;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Objects;
import java.util.concurrent.locks.LockSupport;

/**
 *
 * @author koalton
 */
public class CustomerCenter {

    DatabusConsumerBuilder consumerBuilder = null ;
    public static final SimpleDateFormat Y2MS = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    Logger log = LoggerFactory.getLogger(CustomerCenter.class);
    //两次获取消息的最大时间间隔-60s
    public static final int MAX_WAIT_RECEIVE_GAP_TIME = 60;
    //两次获取消息的初始化时间间隔-1s
    public static final int INIT_RECEIVE_GAP_TIME = 1;
    //两次获取消息时间间隔增长步长 2s
    public static final int RECEIVE_GAP_TIME_STEP = 2;

    public CustomerCenter() {
        try {
            //实例配置对象
            DatabusConfig config = new DatabusConfig("Customer");
//            DatabusConfig config = new DatabusConfig("so_order");
            //通过配置实例化builder对象
            consumerBuilder = new DatabusConsumerBuilder(config);
        } catch (JMSException e1) {
            log.info(e1.toString());
            e1.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
            log.info(e.getMessage());
            e.printStackTrace();
        }
    }

    /**
     * 从服务端获取消息
     */
    @Test
    public void receive() {
        DatabusConsumer consumer = null;
        try {
            //创建消费事件[100001-AddOrder]的消费者,此消费者自动ACK消息 _
//            consumer = this.initConsumer(consumerBuilder,"so_order_committed",1001,null);
            consumer = this.initConsumer(consumerBuilder,"AddOrder",100001,null);
        } catch (JMSException e) {
            e.printStackTrace();
        }
        if( null == consumer ){
            System.out.println("消费者初始化失败");
            return ;
        }
        try {
            int waitTime = INIT_RECEIVE_GAP_TIME;
            while (true) {
                Event event  = consumer.receive(1 * 1000);
                if (event == null) {
                    System.out.println("At " + Y2MS.format(new Date()) + " receive:The Queue is empty");
                    LockSupport.parkUntil(System.currentTimeMillis() + waitTime * 1000);
                    waitTime += RECEIVE_GAP_TIME_STEP;
                    if( waitTime >  MAX_WAIT_RECEIVE_GAP_TIME ){
                        waitTime = MAX_WAIT_RECEIVE_GAP_TIME;
                    }
                    continue;
                }
                waitTime = 1 ;
                System.out.println("At " + Y2MS.format(new Date()) + " receive:" + eventCoreInfo(event));
            }
        } catch (JMSException e) {
            log.info(e.getMessage());
            e.printStackTrace();
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        LockSupport.parkUntil(System.currentTimeMillis() + 30 * 60 * 1000);
    }

    /**
     * 从服务端获取消息时携带消息过滤器
     */
    @Test
    public void receiveWithSelector() {
        DatabusConsumer consumer = null;
        try {
            //创建消费事件[100001-AddOrder]的消费者,此消费者自动ACK消息
            consumer = this.initConsumer(consumerBuilder,"AddOrder",100001,"name1='name1' and name2 =1");
        } catch (JMSException e) {
            e.printStackTrace();
        }
        if( null == consumer ){
            System.out.println("消费者初始化失败");
            return ;
        }
        try {
            int waitTime = INIT_RECEIVE_GAP_TIME;
            while (true) {
                Event event  = consumer.receive(1 * 1000);
                if (event == null) {
                    System.out.println("At " + Y2MS.format(new Date()) + " receive:The Queue is empty");
                    LockSupport.parkUntil(System.currentTimeMillis() + waitTime * 1000);
                    waitTime += RECEIVE_GAP_TIME_STEP;
                    if( waitTime >  MAX_WAIT_RECEIVE_GAP_TIME ){
                        waitTime = MAX_WAIT_RECEIVE_GAP_TIME;
                    }
                    continue;
                }
                waitTime = 1 ;
                System.out.println("At " + Y2MS.format(new Date()) + " receive:" + eventCoreInfo(event));
            }
        } catch (JMSException e) {
            log.info(e.getMessage());
            e.printStackTrace();
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        LockSupport.parkUntil(System.currentTimeMillis() + 30 * 60 * 1000);
    }

    public String eventCoreInfo(Event event) {
        //事件来源
        String eventSource = event.getEventBasicData().getSource();
        //事件序号
        String eventSeq = event.getEventBasicData().getKindId()+"";
        //事件名称
        String eventName = event.getEventBasicData().getName();
        //业务编号
        String businessNo = event.getEventMetaData().getBusinessNo()+"";
        //业务数据
        String centent = event.getEventBusinessData().getContent();
        return "{EventSource:" + eventSource + ",eventSeq:" + eventSeq + ",eventName:" + eventName + ",businessNo:" + businessNo + "}";
    }

    /**
     * 实例消费者
     * @param consumerBuilder 消费者创建者
     * @param eventName 消费者所需消费的事件名称
     * @param eventKindId 消费者所需消费的事件kindId
     * @param consumerSelector 消费者所携带的消息过滤器，如果为NULL表示不会过滤消息
     * @return
     */
    private DatabusConsumer initConsumer(DatabusConsumerBuilder consumerBuilder,String eventName,Integer eventKindId,String consumerSelector) throws JMSException {
        if(Objects.equals(null,consumerBuilder)){
            return null ;
        }
        DatabusConsumer consumer =consumerBuilder.create(eventName,eventKindId,consumerSelector);
        return consumer ;
    }
}
